package com.sduept.component.message.core;

import com.sduept.nwld.dataserver.manager.etl.*;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class RunlogService {
	@Autowired
	private RunLogETLManager rManager;
	@Autowired
	private DailNewequipmentETLManager bstMan;
	@Autowired
	private FaultFeedbackETLManager faultFeedMan;
	@Autowired
	private OperationaRiskETLManager orMan;
	@Autowired
	private OverhaulMsgETLManager oMMan;
	@Autowired
	private ProtectDefectETLManager proMan;
	@Autowired
	private SpecialReportETLManager sRMan;
	@Autowired
	private TZRecordETLManager tzManager;
	@Autowired
	private MainProtectionQuitETLManager quitManager;
	@Autowired
	private OperationaRiskExceedETLManager riskManager;
	@Autowired
	private EquipmentOverhaulDetailETLManager equipmentOverhaulDetailETLManager;

	private ExecutorService executor = Executors.newFixedThreadPool(10);

	@KafkaListener(topics = { "run_log_data" })
	public void onMessage(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
			String msgContent = jo.getString("content");
			JSONObject msgJson = JSONObject.fromObject(msgContent);
			final String datetime = msgJson.getString("updatetime");
			String tableName = msgJson.getString("tableName");
			log.info("接收到消息：run_log_data, 数据表："+tableName+",数据更新时间："+datetime);
			if ("t_daily_runlog".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						rManager.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_newequipment".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						bstMan.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_oms_tzxx".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						tzManager.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_befback".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						faultFeedMan.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_operationalrisks".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						orMan.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_protectoverhaulmag".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						oMMan.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_protect_defect".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						proMan.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_specialreport".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						sRMan.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_mainprotectionquit".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						quitManager.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_daily_protectrisk_exceed".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						riskManager.receiveAndSavePartData(datetime);
					}
				});
			} else if ("t_equipment_overhaul_detail".equals(tableName)) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						equipmentOverhaulDetailETLManager.receiveAndSavePartData(datetime);
					}
				});
			}

		}

	}

}
